use crate::common::*;

use std::collections::{HashMap, HashSet};
use std::str;

use ordered_float::NotNan;
use pyo3::class::basic::PyObjectProtocol;
use pyo3::prelude::*;
use pyo3::types::{PyAny, PyBytes, PyString};
use serde::Deserialize;
use serde_aux::prelude::*;


#[derive(Deserialize)]
struct OkexV3FuturesRecordHeader {
    pub table: String,
    #[serde(default)]
    pub action: String,
}

#[derive(Deserialize)]
struct OkexV3FuturesRecordData {
    pub data: Vec<RecordEntry>,
}


#[derive(Deserialize)]
struct TradeMessage {
    data: Vec<TradeEntry>,
}

#[derive(Deserialize)]
struct TradeEntry {
    side: String,
    trade_id: String,
    #[serde(deserialize_with = "deserialize_number_from_string")]
    price: f64,
    #[serde(deserialize_with = "deserialize_number_from_string")]
    qty: f64,
    instrument_id: String,
    timestamp: String,
}

#[pyclass]
pub struct Tick {
    price: NotNan<f64>,
    qty: f64,
    liquidation_qty: f64,
    order_count: i64,
}

#[pymethods]
impl Tick {
    #[getter]
    fn price(&self) -> f64 {
        self.price.into_inner()
    }
    #[getter]
    fn qty(&self) -> f64 {
        self.qty
    }
    #[getter]
    fn liquidation_qty(&self) -> f64 {
        self.liquidation_qty
    }
    #[getter]
    fn order_count(&self) -> i64 {
        self.order_count
    }
}

#[pyproto]
impl PyObjectProtocol for Tick {
    fn __repr__(&self) -> PyResult<String> {
        Ok(format!("Tick(price={}, qty={})", self.price, self.qty))
    }
}

pub type TickList = Vec<(String, String, String, String)>;

#[derive(Deserialize)]
struct RecordEntry {
    pub instrument_id: String,

    // #[serde(default)]
    pub asks: TickList,

    // #[serde(default)]
    pub bids: TickList,

    pub timestamp: String,
}

extern crate chrono;
use chrono::DateTime;

#[pyclass]
pub struct OkexV3FuturesHandler {
    books: HashMap<String, Py<BookBuilder>>,
    exch_timestamps: HashMap<String, i64>,
    book_callbacks: Vec<PyObject>,
    trade_callbacks: Vec<PyObject>,
    book_symbols: HashSet<String>,
    trade_symbols: HashSet<String>,
    py_fromtimestamp: PyObject,
    py_get_product: PyObject,
}

#[pymethods]
impl OkexV3FuturesHandler {
    #[new]
    fn new_python(obj: &PyRawObject, py: Python) {
        let module = py
            .import("coin.exchange.okex_futures.kr_rest.futures_product")
            .unwrap();
        let cls: PyObject = module.get("OkexFuturesProduct").unwrap().to_object(py);

        obj.init(OkexV3FuturesHandler {
            books: HashMap::new(),
            exch_timestamps: HashMap::new(),
            book_symbols: HashSet::new(),
            book_callbacks: Vec::new(),
            trade_symbols: HashSet::new(),
            trade_callbacks: Vec::new(),
            py_fromtimestamp: py
                .import("datetime")
                .unwrap()
                .get("datetime")
                .unwrap()
                .to_object(py)
                .getattr(py, "fromtimestamp")
                .unwrap(),
            py_get_product: cls.getattr(py, "FromStrNativeProductV3").unwrap(),
        });
    }

    fn on_snapshot(
        &mut self,
        py: Python,
        timestamp: i64,
        record_bytes: &PyBytes,
        _queue_data: &PyAny,
        _topic_data: &PyAny,
    ) -> PyResult<()> {
        let bytes: &[u8] = record_bytes.as_bytes();
        let record = Record::<OkexV3FuturesRecordData> {
            timestamp: timestamp,
            data: serde_json::from_slice(bytes).expect("cannot parse json"),
        };

        let _timestamp = (record.timestamp / 1800_000_000_000_i64 * 1800_000_000_000_i64) as f64
            / 1_000_000_000_f64;
        let _timestamp = self.py_fromtimestamp.call1(py, (_timestamp,))?;

        for entry in &record.data.data {
            let native_symbol: String = entry.instrument_id.clone();

            if !self.book_symbols.contains(&native_symbol) {
                continue;
            }

            let py_book: &mut Py<BookBuilder> = self
                .books
                .entry(native_symbol.clone())
                .or_insert_with(|| Py::new(py, BookBuilder::new()).unwrap());
            {
                let mut book: PyRefMut<BookBuilder> = py_book.as_mut(py);
                book.clear();
                let asks = entry
                    .asks
                    .iter()
                    .map(|(a, b, _c, _d)| (a.parse().unwrap(), b.parse().unwrap()))
                    .collect();
                let bids = entry
                    .bids
                    .iter()
                    .map(|(a, b, _c, _d)| (a.parse().unwrap(), b.parse().unwrap()))
                    .collect();
                book.on_snapshot(record.timestamp, &asks, &bids);
            }

            for callback in &self.book_callbacks {
                callback.call1(py, (native_symbol.clone(), py_book.to_object(py)))?;
            }
        }

        Ok(())
    }

    fn on_diff(
        &mut self,
        py: Python,
        timestamp: i64,
        record_bytes: &PyBytes,
        _queue_data: &PyAny,
        _topic_data: &PyAny,
    ) -> PyResult<()> {
        let bytes: &[u8] = record_bytes.as_bytes();
        let record = Record::<OkexV3FuturesRecordData> {
            timestamp: timestamp,
            data: serde_json::from_slice(bytes).expect("cannot parse json"),
        };

        for entry in &record.data.data {
            let native_symbol: String = entry.instrument_id.clone();

            if !self.book_symbols.contains(&native_symbol) {
                continue;
            }

            if !self.books.contains_key(&native_symbol) {
                continue;
            }

            //  "timestamp": "2019-05-06T07:19:39.348Z"
            //  let rfc3339 = DateTime::parse_from_rfc3339("1996-12-19T16:39:57-08:00")?;

            let exch_timestamp : i64 = DateTime::parse_from_rfc3339(&entry.timestamp).unwrap().timestamp_nanos();

            self.exch_timestamps.entry(native_symbol.clone()).or_insert(0);

            if self.exch_timestamps[&native_symbol] > exch_timestamp {
                continue;
            }
            self.exch_timestamps.insert(native_symbol.clone(), exch_timestamp);

            let py_book: &mut Py<BookBuilder> = self
                .books
                .entry(native_symbol.clone())
                .or_insert_with(|| Py::new(py, BookBuilder::new()).unwrap());
            {
                let mut book: PyRefMut<BookBuilder> = py_book.as_mut(py);
                let asks = entry
                    .asks
                    .iter()
                    .map(|(a, b, _c, _d)| (a.parse().unwrap(), b.parse().unwrap()))
                    .collect();
                let bids = entry
                    .bids
                    .iter()
                    .map(|(a, b, _c, _d)| (a.parse().unwrap(), b.parse().unwrap()))
                    .collect();
                book.on_diff(record.timestamp, &asks, &bids);
            }

            for callback in &self.book_callbacks {
                callback.call1(py, (native_symbol.clone(), py_book.to_object(py)))?;
            }
        }

        Ok(())
    }

    // first book, diff book, trade
    fn on_coin2(
        &mut self,
        py: Python,
        timestamp: i64,
        record_bytes: &PyBytes,
        _queue_data: &PyAny,
        _topic_data: &PyAny,
    ) -> PyResult<()> {
        let bytes: &[u8] = record_bytes.as_bytes();

        let header = serde_json::from_slice::<OkexV3FuturesRecordHeader>(bytes)
            .expect("cannot parse json");

        match &*header.table {
            "futures/depth5" | "futures/depth" => {
                let record = Record::<OkexV3FuturesRecordData> {
                    timestamp: timestamp,
                    data: serde_json::from_slice(bytes).expect("cannot parse json"),
                };

                for entry in &record.data.data {
                    let native_symbol: String = entry.instrument_id.clone();

                    if !self.book_symbols.contains(&native_symbol) {
                        continue;
                    }

                    // create a new book if not exists
                    let py_book: &mut Py<BookBuilder> = self
                        .books
                        .entry(native_symbol.clone())
                        .or_insert_with(|| Py::new(py, BookBuilder::new()).unwrap());
                    {
                        let mut book: PyRefMut<BookBuilder> = py_book.as_mut(py);
                        let asks = entry
                            .asks
                            .iter()
                            .map(|(a, b, _c, _d)| (a.parse().unwrap(), b.parse().unwrap()))
                            .collect();
                        let bids = entry
                            .bids
                            .iter()
                            .map(|(a, b, _c, _d)| (a.parse().unwrap(), b.parse().unwrap()))
                            .collect();

                        match &*header.action {
                            "partial" => book.on_snapshot(record.timestamp, &asks, &bids),
                            // "" is for depth5
                            "" | "insert" | "update" | "delete" => {
                                book.on_diff(record.timestamp, &asks, &bids)
                            }
                            _ => { Err(format!("unknown okex_fut v3 action: {}", header.action)).unwrap() }
                        }
                    }

                    for callback in &self.book_callbacks {
                        callback.call1(py, (native_symbol.clone(), py_book.to_object(py)))?;
                    }
                }

            }
            "futures/trade" => {
                let trades = serde_json::from_slice::<TradeMessage>(bytes)
                    .expect("cannot parse trade json");
                for entry in trades.data {
                    let native_symbol: String = entry.instrument_id.clone();
                    let py_trade: Py<Trade> = Py::new(py, Trade {
                        timestamp: timestamp,
                        price: entry.price,
                        qty: entry.qty,
                        side: entry.side,
                    }).unwrap();
                    for callback in &self.trade_callbacks {
                        callback.call1(py, (native_symbol.clone(), py_trade.to_object(py)))?;
                    }
                }
            }
            _ => {
                // unrelated msg?
                return Ok(());
            }
        }


        Ok(())
    }

    fn on_trade(&mut self, _py: Python) {
        // trade_processor.
        unimplemented!();
    }
    fn register_book_symbol(&mut self, symbol: &PyString) -> PyResult<()> {
        let symbol = symbol.to_string()?.into_owned();
        self.book_symbols.insert(symbol);
        Ok(())
    }
    fn register_book_callback(&mut self, callback: PyObject) {
        self.book_callbacks.push(callback);
    }

    fn register_trade_symbol(&mut self, symbol: &PyString) -> PyResult<()> {
        let symbol = symbol.to_string()?.into_owned();
        self.trade_symbols.insert(symbol);
        Ok(())
    }
    fn register_trade_callback(&mut self, callback: PyObject) {
        self.trade_callbacks.push(callback);
    }
}
